#include <set>
#include <boost/archive/binary_iarchive.hpp>
#include <boost/archive/binary_oarchive.hpp>
#include <boost/serialization/map.hpp>
#include <boost/serialization/string.hpp>
#include "DatasetController.h"
#include "PublicFunction.h"
#include "Define.h"
#include "MetaObject.h"
#include "MetaObjectIO.h"
#include "json/json.h"
#include "StoreObject.h"
#include "StoreObjectIO.h"
#include "Config.h"
#include "ObjectController.h"

using namespace std;

//namespace boost {
//namespace serialization {

//template<class Archive>
//void serialize(Archive & ar, gitstore::DatasetNode & d, const unsigned int version)
//{
//    ar & d.keys;
//}

//} // namespace serialization
//}

namespace gitstore {
CDatasetController::CDatasetController() {
    auto config = CGlobalConfig::getInstance()->getConfigContent();
    auto meta_io_worker = make_shared<CEtcdMetaDataIO>(config["etcd_url"].asString(), AppMode::Dataset);
    _meta_op = make_unique<CObjectController>(meta_io_worker);
//    _key_trans_op = make_unique<CDatasetKeysTrans>();

    switch(static_cast<StoreSystemType>(config["store_type"].asInt())) {
    case StoreSystemType::Ceph:
        _store_io_worker = make_unique<CCephObjectIO>(config["ceph_cluster_name"].asString(), config["ceph_user_name"].asString(), config["ceph_conf_path"].asString(), config["ceph_object_pool_name"].asString());
        break;
    default:
        _store_io_worker = make_unique<CLinuxObjectIOBase>();
        break;
    }
}

CDatasetController::~CDatasetController() {

}

ErrorStatus CDatasetController::listDataset(std::vector<std::string>& vds) {
    return _meta_op->listObject(vds);
}

ErrorStatus CDatasetController::createDataset(const std::string& ds_name, const std::string& ds_note,
                                              const std::string& br_name, const std::string& br_note,
                                              StoreObjectType type, const std::vector<std::string>& vks,
                                              const std::vector<StoreObject>& vos, int& revision_ds,
                                              int& revision_branch) {
    if ( type == StoreObjectType::List || type == StoreObjectType::Map ) {
        return SubTypeOnlyCanBeStringOrBlob;
    }
    StoreObject ds_node(StoreObjectType::Map);
    for ( auto i = 0UL; i<vks.size() ; i++ ) {
        auto sha = sha256(vos[i].getSerializedValue());
        _store_io_worker->setValue(sha, vos[i]);
        ds_node[vks[i]] = sha;
    }
    auto res = _meta_op->createObject(ds_name, ds_note, StoreObjectType::Map, type, br_name, br_note,
                                      "init", ds_node, revision_ds, revision_branch);
    return res;
}

ErrorStatus CDatasetController::listDatasetBranches(const std::string& ds_name, std::vector<std::string>& vbr) {
    return _meta_op->listObjectBranches(ds_name, vbr);
}

ErrorStatus CDatasetController::branchDataset(const std::string& ds_name, const std::string& br_refer_name,
                                              const std::string& br_new_name, const std::string& br_note,
                                              int revision_refer_branch, int& revision_new_branch) {
    auto res = _meta_op->branchObject(ds_name, br_refer_name, br_new_name,
                                      br_note, revision_refer_branch, revision_new_branch);
    return res;
}

ErrorStatus CDatasetController::listDatasetKeysByBranch(const std::string& ds_name, const std::string& br_name,
                                                        std::vector<std::string>& vbk, int& revision_branch) {
    StoreObject storeobj(StoreObjectType::Map);
    auto res = _meta_op->getValueByBranch(ds_name, br_name, storeobj, revision_branch);
    if ( Success != res ) {
        return res;
    }
    vbk.clear();
    vbk.shrink_to_fit();
    vbk.reserve(storeobj.size());
    for ( auto iter = storeobj.mbegin() ; iter != storeobj.mend() ; iter++ ) {
        vbk.emplace_back(iter->first);
    }
    return res;
}

ErrorStatus CDatasetController::listDatasetKeysByVersion(const std::string& ds_name, const std::string& version,
                                                         std::vector<std::string>& vbk) {
    StoreObject storeobj(StoreObjectType::Map);
    auto res = _meta_op->getValueByVersion(ds_name, version, storeobj);
    if ( Success != res ) {
        return res;
    }
    vbk.clear();
    vbk.shrink_to_fit();
    vbk.reserve(storeobj.size());
    for ( auto iter = storeobj.mbegin() ; iter != storeobj.mend() ; iter++ ) {
        vbk.emplace_back(iter->first);
    }
    return res;
}

ErrorStatus CDatasetController::putKey(const std::string& ds_name, const std::string& br_name, const std::string& key,
                                       const StoreObject& value, int revision_pre, int& revision_new) {
    StoreObject storeobj(StoreObjectType::Map);
    int revision_branch_cur = NO_REVISION;
    auto res = _meta_op->getValueByBranch(ds_name, br_name, storeobj, revision_branch_cur);
    if ( Success != res ) {
        return res;
    }

    auto obj_hash = sha256(value.getSerializedValue());
    if (( storeobj.isKeyExist(key) )
            && ( storeobj[key].compare(obj_hash) == 0 )) {
        return ValueNoChange;
    }
    res = _store_io_worker->setValue(obj_hash, value);
    if ( Success != res ) {
        return res;
    }
    storeobj[key] = obj_hash;

    res = _meta_op->putValue(ds_name, br_name, storeobj,
                             "", revision_pre, revision_new);
    return res;
}

ErrorStatus CDatasetController::getKeyByBranch(const std::string& ds_name, const std::string& br_name,
                                               const std::string& key, StoreObject& value, int& revision_branch) {
    StoreObject storeobj(StoreObjectType::Map);
    auto res = _meta_op->getValueByBranch(ds_name, br_name, storeobj, revision_branch);
    if ( Success != res ) {
        return res;
    }
    if ( storeobj.isKeyExist(key) ) {
        return KeyNotExist;
    }
    res = _store_io_worker->getValue(storeobj[key], value);
    return res;
}

ErrorStatus CDatasetController::getKeyByVersion(const std::string& ds_name, const std::string& version,
                                                const std::string& key, StoreObject& value) {
    StoreObject storeobj(StoreObjectType::Map);
    auto res = _meta_op->getValueByVersion(ds_name, version, storeobj);
    if ( Success != res ) {
        return res;
    }
    res = _store_io_worker->getValue(storeobj[key], value);
    return res;
}

ErrorStatus CDatasetController::getDiffKeysBranch(const std::string& ds_name, const std::string& br_name1, const std::string& br_name2,
                                                  std::vector<std::string>& vks, int& revision_branch1, int& revision_branch2) {
    return _meta_op->diffObjectBranch(ds_name, br_name1, ds_name, br_name2, revision_branch1, revision_branch2, vks);

//    Object object;
//    int revision_dataset = NO_REVISION;
//    auto res = _meta_op->getObjectInfo(ds_name, object, revision_dataset);
//    if ( Success != res ) {
//        return res;
//    }
//    StoreObject value1(StoreObjectType::Map), value2(StoreObjectType::Map);
//    res = _meta_op->getValueByBranch(object, br_name1, value1, revision_branch1);
//    if ( Success != res ) {
//        return res;
//    }
//    res = _meta_op->getValueByBranch(object, br_name2, value2, revision_branch2);
//    if ( Success != res ) {
//        return res;
//    }
//    return findDiffKeys(value1, value2, vks);
}

ErrorStatus CDatasetController::getDiffKeysVersion(const std::string& ds_name, const std::string& version1, const std::string& version2, std::vector<std::string>& vks) {
    return _meta_op->diffObjectVersion(ds_name, version1, ds_name, version2, vks);

//    Object object;
//    int revision_dataset = NO_REVISION;
//    auto res = _meta_op->getObjectInfo(ds_name, object, revision_dataset);
//    if ( Success != res ) {
//        return res;
//    }
//    StoreObject value1(StoreObjectType::Map), value2(StoreObjectType::Map);
//    res = _meta_op->getValueByVersion(object, version1, value1);
//    if ( Success != res ) {
//        return res;
//    }
//    res = _meta_op->getValueByVersion(object, version2, value2);
//    if ( Success != res ) {
//        return res;
//    }
//    return findDiffKeys(value1, value2, vks);
}

//ErrorStatus CDatasetController::findDiffKeys(const StoreObject& value1, const StoreObject& value2, std::vector<std::string>& vks) {
//    //    Json::Reader reader;
//    //    Json::Value json_value1,json_value2;
//    //    reader.parse(value1.getData(), json_value1);
//    //    reader.parse(value2.getData(), json_value2);

////    const DatasetNode *ds_node1 = value1.getValue();
////    const DatasetNode *ds_node2 = value2.getValue();
//    set<string> diff_set;
//    for ( auto iter = value1.mbegin() ; iter != value1.mend() ; iter++ ) {
//        auto key = iter->first;
//        if ( value2.isKeyExist(key) ) {
//            diff_set.emplace(key);
//            continue;
//        } else {
//            if ( iter->second.compare(value2.get(key)) != 0 ) {
//                diff_set.emplace(key);
//            }
//        }
//    }
//    for ( auto iter = value2.mbegin() ; iter != value2.mend() ; iter++ ) {
//        auto key = iter->first;
//        if ( diff_set.find(key) != diff_set.end() )
//            continue;
//        if ( value1.isKeyExist(key) ) {
//            diff_set.emplace(key);
//            continue;
//        } else {
//            if ( iter->second.compare(value1.get(key)) != 0 ) {
//                diff_set.emplace(key);
//            }
//        }
//    }
//    vks.clear();
//    vks.reserve(diff_set.size());
//    vks.insert(vks.begin(), diff_set.begin(), diff_set.end());
//    return Success;
//}

ErrorStatus CDatasetController::getBranchInfo(const std::string& ds_name, const std::string& br_name, Branch& branch, int& revision_branch) {
    return _meta_op->getBranchInfo(ds_name, br_name, branch, revision_branch);
}



//CDatasetKeysTrans::CDatasetKeysTrans() {

//}

//CDatasetKeysTrans::~CDatasetKeysTrans() {

//}

//ErrorStatus CDatasetKeysTrans::encode(const DatasetNode& keys, std::string& str) {
//    std::ostringstream ss;
//    boost::archive::binary_oarchive ob(ss);
//    ob << keys;
//    str.assign(ss.str());
//    return Success;
//}

//ErrorStatus CDatasetKeysTrans::decode(const std::string& str, DatasetNode& keys) {
//    std::istringstream ss(str);
//    boost::archive::binary_iarchive ob(ss);
//    ob >> keys;
//    return Success;
//}

}
